package com.innodealing.source;

import com.innodealing.model.User;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 简化的用户快照数据源 - 基于 Table API 和 MySQL JDBC 连接器
 *
 * @author 系统生成
 * @date 2024/12/18
 */
public class TableApiUserSnapshotSource {

    private static final Logger logger = LoggerFactory.getLogger(TableApiUserSnapshotSource.class);

    private final String jdbcUrl;
    private final String username;
    private final String password;
    private final String tableName;

    /**
     * 构造函数
     *
     * @param jdbcUrl   JDBC 连接 URL
     * @param username  数据库用户名
     * @param password  数据库密码
     * @param tableName 表名（可选，默认为 "user"）
     */
    public TableApiUserSnapshotSource(String jdbcUrl, String username, String password, String tableName) {
        this.jdbcUrl = jdbcUrl;
        this.username = username;
        this.password = password;
        this.tableName = tableName != null ? tableName : "user";

        logger.info("创建TableApiUserSnapshotSource实例: url={}, tableName={}",
                jdbcUrl, this.tableName);
    }

    /**
     * 创建用户快照数据流
     *
     * @param env Flink 流环境
     * @return 用户数据流
     */
    public DataStream<User> createStream(StreamExecutionEnvironment env) {
        try {
            logger.info("开始创建用户快照数据流 - 这表明正在进行快照数据查询");
            logger.info("连接信息: URL={}, 表名={}", jdbcUrl, tableName);

            // 创建 Table Environment
            StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

            // 注册 MySQL 表
            logger.info("开始注册MySQL表...");
            registerMysqlTable(tableEnv);
            logger.info("MySQL表注册完成");

            // 查询并转换为 DataStream
            logger.info("开始执行SQL查询: SELECT * FROM `{}`", tableName);
            Table userTable = tableEnv.sqlQuery("SELECT * FROM `" + tableName + "`");

            logger.info("开始将Table转换为DataStream...");
            DataStream<User> resultStream = tableEnv.toDataStream(userTable)
                    .map(new UserRowMapper())
                    .returns(TypeInformation.of(User.class));

            logger.info("用户快照数据流创建完成！这意味着快照数据将被读取");
            return resultStream;

        } catch (Exception e) {
            logger.error("创建用户数据流失败: {}", e.getMessage(), e);
            throw new RuntimeException("创建用户数据流失败", e);
        }
    }

    /**
     * 注册 MySQL 表
     */
    private void registerMysqlTable(StreamTableEnvironment tableEnv) {
        String ddl = String.format(
                "CREATE TABLE `%s` (" +
                        "  id INT," +
                        "  username STRING," +
                        "  email STRING," +
                        "  age INT," +
                        "  create_time TIMESTAMP(3)," +
                        "  update_time TIMESTAMP(3)" +
                        ") WITH (" +
                        "  'connector' = 'jdbc'," +
                        "  'url' = '%s'," +
                        "  'table-name' = '%s'," +
                        "  'username' = '%s'," +
                        "  'password' = '%s'," +
                        "  'driver' = 'com.mysql.cj.jdbc.Driver'," +
                        "  'scan.partition.column' = 'id'," +
                        "  'scan.partition.num' = '4'," +
                        "  'scan.partition.lower-bound' = '1'," +
                        "  'scan.partition.upper-bound' = '10000'" +
                        ")",
                tableName, jdbcUrl, tableName, username, password
        );

        logger.info("执行DDL语句: {}", ddl);
        tableEnv.executeSql(ddl);
        logger.info("DDL执行完成");
    }

    /**
     * Row 到 User 的映射器
     */
    private static class UserRowMapper implements MapFunction<Row, User> {
        private static final Logger logger = LoggerFactory.getLogger(UserRowMapper.class);

        @Override
        public User map(Row row) {
            User user = new User();
            user.setId((Integer) row.getField(0));
            user.setUsername((String) row.getField(1));
            logger.info("转换用户记录: id={}, username={}", user.getId(), user.getUsername());
            return user;
        }
    }

    // ==================== 静态工厂方法 ====================

    /**
     * 快速创建用户数据流（使用默认表名 "user"）
     */
    public static DataStream<User> fromMysql(StreamExecutionEnvironment env,
                                             String jdbcUrl, String username, String password) {
        logger.info("通过fromMysql静态方法创建快照数据流 (默认表名)");
        return new TableApiUserSnapshotSource(jdbcUrl, username, password, null)
                .createStream(env);
    }

    /**
     * 快速创建用户数据流（指定表名）
     */
    public static DataStream<User> fromMysql(StreamExecutionEnvironment env,
                                             String jdbcUrl, String username, String password, String tableName) {
        logger.info("通过fromMysql静态方法创建快照数据流 (表名: {})", tableName);
        return new TableApiUserSnapshotSource(jdbcUrl, username, password, tableName)
                .createStream(env);
    }

    /**
     * 从数据库配置创建数据流
     */
    public static DataStream<User> fromMysql(StreamExecutionEnvironment env,
                                             String host, int port, String database,
                                             String username, String password, String tableName) {
        String jdbcUrl = String.format("jdbc:mysql://%s:%d/%s", host, port, database);
        logger.info("通过主机配置创建快照数据流: {}:{}/{}", host, port, database);
        return fromMysql(env, jdbcUrl, username, password, tableName);
    }
} 